package com.learning.hadoop.mapreduce.join;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;

/**
 * map端join算法实现
 * 适用于关联表中有小表的情形；
 * 可以将小表分发到所有的map节点，这样，map节点就可以在本地对自己所读到的大表数据进行join并输出最终结果，可以大大提高join操作的并发度，加快处理速度
 * <p>
 * –先在mapper类中预先定义好小表，进行join
 * –引入实际场景中的解决方案：一次加载数据库或者用distributedcache
 *
 * @author Sam Sho
 */
public class MapSideJoin {

    static class MapSideJoinMapper extends Mapper<LongWritable, Text, InfoBean, NullWritable> {
        Map<String, InfoBean> pdInfoMap = new HashMap<String, InfoBean>();

        InfoBean bean = new InfoBean();

        /**
         * 通过阅读父类Mapper的源码，发现 setup方法是在maptask处理数据之前调用一次 可以用来做一些初始化工作
         */
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("product.txt")));
            String line;

            while (StringUtils.isNotEmpty(line = br.readLine())) {
                InfoBean pdBean = new InfoBean();
                String[] fields = line.split("\t");
                pdBean.set("", "", fields[0], -1, fields[1], fields[2], Float.parseFloat(fields[3]), "1");
                pdInfoMap.put(fields[0], pdBean);
            }
            br.close();
        }

        // 由于已经持有完整的产品信息表，所以在map方法中就能实现join逻辑了
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");
            String pid = fields[2];
            //InfoBean productBean = pdInfoMap.get(pid);
            bean.setOrderId(fields[0]);
            bean.setDate(fields[1]);
            bean.setPid(pid);
            bean.setAmount(Integer.parseInt(fields[3]));
            bean.setPName(pdInfoMap.get(pid).getPName());
            bean.setCategoryId(pdInfoMap.get(pid).getCategoryId());
            bean.setPrice(pdInfoMap.get(pid).getPrice());
            context.write(bean, NullWritable.get());
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        // 指定本程序的jar包所在的本地路径
        job.setJarByClass(MapSideJoin.class);

        //System.setProperty("hadoop.home.dir", "D:\\hadoop-2.6.5");

        // 指定本业务job要使用的mapper/Reducer业务类
        job.setMapperClass(MapSideJoinMapper.class);

        // 指定mapper输出数据的kv类型
        job.setMapOutputKeyClass(InfoBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //FileInputFormat.setInputPaths(job, new Path("hdfs://mini1/mapsidejoin/input"));
        //FileOutputFormat.setOutputPath(job, new Path("hdfs://mini1/mapsidejoin/output"));

        // 指定需要缓存一个文件到所有的maptask运行节点工作目录
        /* job.addArchiveToClassPath(archive); */// 缓存jar包到task运行节点的classpath中
        /* job.addFileToClassPath(file); */// 缓存普通文件到task运行节点的classpath中
        /* job.addCacheArchive(uri); */// 缓存压缩包文件到task运行节点的工作目录
        /* job.addCacheFile(uri) */// 缓存普通文件到task运行节点的工作目录

        // 将产品表文件缓存到task工作节点的工作目录中去
        job.addCacheFile(new URI("hdfs://mini1/mapsidejoin/cache/product.txt"));

        // map端join的逻辑不需要reduce阶段，设置reducetask数量为0
        job.setNumReduceTasks(0);

        boolean res = job.waitForCompletion(true);
        System.exit(res ? 0 : 1);
    }

}